package com.data.capture.converter;

import io.debezium.spi.converter.CustomConverter;
import io.debezium.spi.converter.RelationalColumn;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.connect.data.SchemaBuilder;

import java.time.Duration;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.time.ZoneId;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Objects;
import java.util.Properties;

/**
 * MySQL 日期字段时区/格式处理
 * @author linch
 **/
public class MySqlDateTimeConverter implements CustomConverter<SchemaBuilder, RelationalColumn> {

    @Override
    public void configure(Properties props) {
    }

    // 注册转换器
    @Override
    public void converterFor(RelationalColumn column, ConverterRegistration<SchemaBuilder> registration) {
        String type = column.typeName().toUpperCase();
        if (StringUtils.isBlank(type)) {
            return;
        }
        switch (column.typeName().toUpperCase()) {
            case "DATE":
                registration.register(
                    SchemaBuilder.string().optional().name("com.darcytech.debezium.date.string"),
                    this::convertDate
                );
                break;
            case "TIME":
                registration.register(
                    SchemaBuilder.string().optional().name("com.darcytech.debezium.time.string"),
                    this::convertTime
                );
                break;
            case "DATETIME":
                registration.register(
                    SchemaBuilder.string().optional().name("com.darcytech.debezium.datetime.string"),
                    this::convertDateTime
                );
                break;
            case "TIMESTAMP":
                registration.register(
                    SchemaBuilder.string().optional().name("com.darcytech.debezium.timestamp.string"),
                    this::convertTimestamp
                );
                break;
            default:
                break;
        }
    }

    // 转换日期
    private String convertDate(Object input) {
        if (Objects.isNull(input)) {
            return null;
        }
        if (input instanceof LocalDate) {
            return DateTimeFormatter.ISO_DATE.format((LocalDate) input);
        }
        if (input instanceof Integer) {
            LocalDate date = LocalDate.ofEpochDay((Integer) input);
            return DateTimeFormatter.ISO_DATE.format(date);
        }
        return String.valueOf(input);
    }

    // 转换时间
    private String convertTime(Object input) {
        if (Objects.isNull(input)) {
            return null;
        }
        if (input instanceof Duration) {
            Duration duration = (Duration) input;
            return DateTimeFormatter.ISO_TIME.format(
                LocalTime.ofSecondOfDay(duration.getSeconds()).withNano(duration.getNano())
            );
        }
        return String.valueOf(input);
    }

    // 转换日期时间
    private String convertDateTime(Object input) {
        if (Objects.isNull(input)) {
            return null;
        }
        if (input instanceof LocalDateTime) {
            return DateTimeFormatter.ISO_DATE_TIME.format((LocalDateTime) input).replaceAll("T", " ");
        }
        return String.valueOf(input);
    }

    // 转换时间戳
    private String convertTimestamp(Object input) {
        if (Objects.isNull(input)) {
            return null;
        }
        if (input instanceof ZonedDateTime) {
            // mysql的timestamp会转成UTC存储，这里的zonedDatetime都是UTC时间
            ZonedDateTime zonedDateTime = (ZonedDateTime) input;
            LocalDateTime localDateTime = zonedDateTime.withZoneSameInstant(ZoneId.systemDefault()).toLocalDateTime();
            return DateTimeFormatter.ISO_DATE_TIME.format(localDateTime).replaceAll("T", " ");
        }
        return String.valueOf(input);
    }
}
